package com.taxi.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.taxi.dto.OrderInfo;
import com.taxi.service.DispatchService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.concurrent.*;

@Component
@Slf4j
public class KafkaConsumer {
    @Autowired
    private DispatchService dispatchService;

    @Autowired
    private KafkaProducer kafkaProducer;

    /**
     * 高度 io 任务，核心线程数设置为cpu核数的2倍
     */
    private int coreSize = 2*Runtime.getRuntime().availableProcessors();

    private final ExecutorService executorService = Executors.newFixedThreadPool(coreSize);
    /**
     * 消费者
     * @param records
     * @param acknowledgment
     * @return
     * @describute concurrency = "2" 表示开启两个消费者线程消费两个分区
     */
    @KafkaListener(topics = {"travel-topic"}, concurrency = "2")
    public void consumer(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        // 倒计时锁，取一半，防止线程池溢出，当不减少时，说明出现了严重的司乘供需不平衡，此时不再从kafka马上拉取
        // 直到一半任务完成，说明线程池有一半空余线程了，可以继续拉取订单
        CountDownLatch latch = new CountDownLatch(records.size()/2);
        for (ConsumerRecord<String, String> record : records) {
            executorService.execute(() -> {
                log.info("consumer, topic:{}, partition:{}, value: {}",
                        record.topic(), record.partition(), record.value());
                // 消费消息，订单匹配
                consumeMsg(record.value());
                latch.countDown(); // 任务提交，计数器减一
            });
        }

        try {
            latch.await(); // 等待一半任务执行完成就可以提交拉取下一批了，说明至少空出了一半线程
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 手动异步提交偏移量
        acknowledgment.acknowledge();
    }

    /**
     * 消费消息
     * @param msg
     */
    public void consumeMsg(String msg){
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            OrderInfo orderInfo = objectMapper.readValue(msg, OrderInfo.class);
            // 订单匹配
            dispatchService.disPatch(orderInfo);
        } catch (Exception e) {
            e.printStackTrace();
            // 将消费失败的消息放到另一个专门处理失败消息的主题 "errOrderHandle-topic"
            kafkaProducer.sendMessage("errOrderHandle-topic",msg);
        }
    }
}
